const amqp = require('amqplib');

class RabbitMQ {
  constructor() {
    this.url = 'amqp://panda:Yb222341@120.26.120.135:5672';
    this.queueName = 'defaultQueue';
    this.connection = null;
    this.channel = null;
    this.consumers = new Map();
  }

  async connect() {
    try {
      this.connection = await amqp.connect(this.url);
      this.channel = await this.connection.createChannel();
      console.log('Connected to RabbitMQ Successfully');
    } catch (error) {
      console.error('Connected to RabbitMQ Error');
      throw error;
    }
  }

  async sendMessage(message) {
    return new Promise(async (resolve, reject) => {
      if (!this.channel) {
        reject(new Error('RabbitMQ channel is not initialized.'));
        return;
      }
      try {
        await this.channel.assertQueue(this.queueName, { durable: true });
        await this.channel.sendToQueue(this.queueName, Buffer.from(message));
        console.log(`Message sent to queue ${this.queueName}: ${message}`);
        resolve(true);
      } catch (error) {
        console.log('Failed to send message:', error);
        reject(error);
      }
    });
  }

  async consumeMessages() {
    console.log('Waiting for messages...');
    if (!this.channel) {
      throw new Error('RabbitMQ channel is not initialized.');
    }

    // 如果已存在消费者，先取消它
    if (this.consumers.has(this.queueName)) {
      const consumerTag = this.consumers.get(this.queueName);
      await this.channel.cancel(consumerTag);
      this.consumers.delete(this.queueName);
    }

    return new Promise(async (resolve, reject) => {
      try {
        await this.channel.assertQueue(this.queueName, { durable: true });
        const { consumerTag } = await this.channel.consume(this.queueName, async msg => {
          if (!msg) {
            return;
          }

          try {
            console.log('Received message:', msg.content.toString());
            const res = JSON.parse(msg.content.toString());
            if (res) {
              this.channel.ack(msg);
              resolve(res);
            } else {
              this.channel.nack(msg, false, true);
              reject(new Error('Invalid message format'));
            }
          } catch (error) {
            console.error('Error processing message:', error);
            this.channel.nack(msg, false, true);
            reject(error);
          }
        }, { noAck: false });

        this.consumers.set(this.queueName, consumerTag);
      } catch (error) {
        console.error('Error setting up consumer:', error);
        reject(error);
      }
    });
  }
}


module.exports = RabbitMQ;
